/**
 * 基于窗口join
 */
package com.atguigu.day5

import org.apache.flink.api.common.functions.JoinFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object WindowJoinExceple {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //    设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val input1 = env.fromElements(
      ("a", 1, 1000L),
      ("a", 2, 2000L),
      ("b", 1, 3000L),
      ("b", 2, 4000L)
    ).assignAscendingTimestamps(_._3)


    val input2 = env.fromElements(
      ("a", 10, 1000L),
      ("a", 20, 2000L),
      ("b", 10, 3000L),
      ("b", 20, 4000L)
    ).assignAscendingTimestamps(_._3)

    input1
      .join(input2)
      // on  input1.1=input2.1
      .where(_._1)
      .equalTo(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      .apply(new MyJoinFunction)
      .print()
    env.execute()




  }
  //分流开窗口以后，属于同一窗口的input1中的元素和input2中的元素做笛卡尔积
  //相同的key ，而且是相同的窗口，中的元素做笛卡尔积
  class MyJoinFunction extends JoinFunction[(String,Int,Long),(String,Int,Long),String]{
    override def join(in1: (String, Int, Long), in2: (String, Int, Long)): String = {
      //不用out 直接就可输出
      in1+"===========>"+in2
    }
  }
}
